package DyingBaby.backend.plan;

import DyingBaby.backend.catalog.Catalog;
import DyingBaby.backend.catalog.TableSchema;
import DyingBaby.backend.dm.DataManager;
import DyingBaby.backend.sql.Ast;
import DyingBaby.backend.table.PagedTableHeap;
import DyingBaby.backend.table.TableHeap;
import DyingBaby.backend.tm.TransactionManager;

import java.io.File;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Predicate;

/**
 * 逻辑执行器 (LogicalExecutor)
 * 主要功能：
 * 1. 接收逻辑计划 (LogicalPlan)，并根据计划类型分发给具体的执行方法
 * 2. 实现各种SQL操作的实际执行逻辑
 * 3. 管理事务和数据存储的交互
 * 4. 支持两种存储后端：分页文件存储和JSON堆存储
 */
public class LogicalExecutor {
    // 数据库目录（元数据管理器），用于获取表结构信息
    private final Catalog catalog;

    // 构造函数，接收数据库目录
    public LogicalExecutor(Catalog catalog) { this.catalog = catalog; }

    // 判断是否使用分页存储（基于系统配置）
    private boolean usePaged() { return Boolean.parseBoolean(System.getProperty("mydb.paged", "true")); }

    /**
     * 执行器的主要入口方法
     * 根据逻辑计划类型分发到具体的执行方法
     * @param plan 逻辑执行计划
     * @return 执行结果（可能是操作状态、查询结果集或影响的行数）
     */
    public Object execute(LogicalPlan plan) {
        if (plan instanceof CreateTablePlan) return execCreate((CreateTablePlan) plan);
        if (plan instanceof InsertPlan) return execInsert((InsertPlan) plan);
        if (plan instanceof ProjectPlan) return execProject((ProjectPlan) plan);
        if (plan instanceof FilterPlan) return execFilter((FilterPlan) plan);
        if (plan instanceof SeqScanPlan) return execSeqScan((SeqScanPlan) plan);
        if (plan instanceof PkLookupPlan) return execPkLookup((PkLookupPlan) plan);
        if (plan instanceof UpdatePlan) return execUpdate((UpdatePlan) plan);
        if (plan instanceof DeletePlan) return execDelete((DeletePlan) plan);
        throw new IllegalArgumentException("Unsupported logical plan: " + plan.getClass().getSimpleName());
    }

    /**
     * 执行创建表操作
     * @param p 创建表计划
     * @return 操作结果
     */
    private Object execCreate(CreateTablePlan p) {
        TableSchema schema = new TableSchema(p.table, p.columns, p.primaryKey);
        catalog.createTable(schema);
        return "Successful!";
    }

    /**
     * 执行插入数据操作
     * 支持两种存储后端：分页文件存储和JSON堆存储
     * @param p 插入计划
     * @return 操作结果
     */
    private Object execInsert(InsertPlan p) {
        // 获取表结构信息
        TableSchema schema = catalog.getTable(p.table);
        if (usePaged()) {
            // 使用分页文件存储（更接近真实数据库的存储方式）
            String tmBase = "data/" + p.table + "_tm";
            String tmFilePath = tmBase + ".xid";

            // 判断事务管理器文件是否已存在
            boolean tmExists = new File(tmFilePath).exists();

            // 打开或创建事务管理器
            TransactionManager tm = tmExists ? TransactionManager.open(tmBase) : TransactionManager.create(tmBase);

            // 打开或创建数据管理器
            DataManager dm = tmExists ? DataManager.open("data/" + p.table + "_dm", 1<<20, tm)
                    : DataManager.create("data/" + p.table + "_dm", 1<<20, tm);

            // 创建分页表堆
            PagedTableHeap heap = new PagedTableHeap(schema, tm, dm);

            // 执行插入操作
            heap.insert(new HashMap<>(p.values));

            // 关闭资源
            dm.close(); tm.close();
        } else {
            // 使用JSON堆存储（更简单的存储方式，用于测试或简单场景）
            TableHeap heap = new TableHeap(schema);
            heap.insert(new HashMap<>(p.values));
        }
        return "Successful!";
    }

    /**
     * 执行顺序扫描操作
     * 读取表中的所有数据
     * @param p 顺序扫描计划
     * @return 查询结果（数据行列表）
     */
    private Object execSeqScan(SeqScanPlan p) {
        // 获取表结构信息
        TableSchema schema = catalog.getTable(p.table);
        List<Map<String,Object>> rows;

        if (usePaged()) {
            // 使用分页文件存储
            TransactionManager tm = TransactionManager.open("data/" + p.table + "_tm");
            DataManager dm = DataManager.open("data/" + p.table + "_dm", 1<<20, tm);
            PagedTableHeap heap = new PagedTableHeap(schema, tm, dm);

            // 执行扫描操作（null表示无过滤条件）
            rows = heap.scan(null);

            // 关闭资源
            dm.close(); tm.close();
        } else {
            TableHeap heap = new TableHeap(schema);
            rows = heap.scan(null);
        }
        return rows;
    }

    //执行优化后的路径
    private Object execPkLookup(PkLookupPlan p) {
        TableSchema schema = catalog.getTable(p.table);
        List<Map<String,Object>> rows = new ArrayList<>();
        if (usePaged()) {
            // 1. 打开事务和数据管理器（访问磁盘文件的入口）
            TransactionManager tm = TransactionManager.open("data/" + p.table + "_tm");
            DataManager dm = DataManager.open("data/" + p.table + "_dm", 1<<20, tm);

            // 2. 创建或获取表的堆结构
            PagedTableHeap heap = new PagedTableHeap(schema, tm, dm);

            // 3. 通过主键值获取记录(我现在的核心优化)
            Map<String,Object> rec = heap.getByPrimaryKey(p.keyValue);
            if (rec != null) rows.add(rec);
            dm.close(); tm.close();
        } else {
            // JSON heap 没有索引，退化为全表扫描
            TableHeap heap = new TableHeap(schema);
            for (Map<String,Object> r : heap.scan(null)) {
                String pk = schema.getPrimaryKey();
                if (pk != null) {
                    Object v = r.get(pk);
                    if (p.keyValue == null ? v == null : p.keyValue.equals(v)) rows.add(r);
                }
            }
        }
        return rows;
    }

    /**
     * 执行过滤操作
     * 对输入数据应用过滤条件，只保留满足条件的行
     * @param p 过滤计划
     * @return 过滤后的数据
     */
    private Object execFilter(FilterPlan p) {
        // 递归执行子计划（通常是SeqScanPlan）获取输入数据
        @SuppressWarnings("unchecked")
        List<Map<String,Object>> input = (List<Map<String, Object>>) execute(p.child);

        // 准备输出列表
        List<Map<String,Object>> out = new ArrayList<>();

        // 应用过滤条件
        for (Map<String,Object> row : input) {
            if (evaluateExpression(p.predicate, row)) {
                out.add(row);
            }
        }
        return out;
    }

    // 评估表达式的值
    private boolean evaluateExpression(Ast.Expr expr, Map<String, Object> row) {
        if (expr instanceof Ast.Eq) {
            Ast.Eq eq = (Ast.Eq) expr;
            Object v = row.get(eq.col);
            return eq.value == null ? v == null : eq.value.equals(v);
        } else if (expr instanceof Ast.Ne) {
            Ast.Ne ne = (Ast.Ne) expr;
            Object v = row.get(ne.col);
            return ne.value == null ? v != null : !ne.value.equals(v);
        } else if (expr instanceof Ast.Gt) {
            Ast.Gt gt = (Ast.Gt) expr;
            Object v = row.get(gt.col);
            return compareValues(v, gt.value) > 0;
        } else if (expr instanceof Ast.Lt) {
            Ast.Lt lt = (Ast.Lt) expr;
            Object v = row.get(lt.col);
            return compareValues(v, lt.value) < 0;
        } else if (expr instanceof Ast.Ge) {
            Ast.Ge ge = (Ast.Ge) expr;
            Object v = row.get(ge.col);
            return compareValues(v, ge.value) >= 0;
        } else if (expr instanceof Ast.Le) {
            Ast.Le le = (Ast.Le) expr;
            Object v = row.get(le.col);
            return compareValues(v, le.value) <= 0;
        } else if (expr instanceof Ast.And) {
            Ast.And and = (Ast.And) expr;
            return evaluateExpression(and.left, row) && evaluateExpression(and.right, row);
        } else if (expr instanceof Ast.Or) {
            Ast.Or or = (Ast.Or) expr;
            return evaluateExpression(or.left, row) || evaluateExpression(or.right, row);
        } else if (expr instanceof Ast.Not) {
            Ast.Not not = (Ast.Not) expr;
            return !evaluateExpression(not.expr, row);
        } else if (expr instanceof Ast.Paren) {
            Ast.Paren paren = (Ast.Paren) expr;
            return evaluateExpression(paren.expr, row);
        } else {
            throw new IllegalArgumentException("Unsupported expression type: " + expr.getClass().getSimpleName());
        }
    }

    // 比较两个值的大小
    private int compareValues(Object v1, Object v2) {
        if (v1 == null && v2 == null) return 0;
        if (v1 == null) return -1;
        if (v2 == null) return 1;

        // 如果都是数字，进行数值比较
        if (v1 instanceof Number && v2 instanceof Number) {
            double d1 = ((Number) v1).doubleValue();
            double d2 = ((Number) v2).doubleValue();
            return Double.compare(d1, d2);
        }

        // 否则进行字符串比较
        return v1.toString().compareTo(v2.toString());
    }


    /**
     * 执行投影操作
     * 目前实现是选择所有列（相当于无操作）
     * @param p 投影计划
     * @return 投影后的数据
     */
    private Object execProject(ProjectPlan p) {
        return execute(p.child); // project all columns for now
    }

    /**
     * 执行更新操作
     * 目前只支持分页存储后端
     * @param p 更新计划
     * @return 更新影响的行数
     */
    private Object execUpdate(UpdatePlan p) {
        if (!usePaged()) return "UPDATE not implemented for JSON heap";

        // 获取表结构信息
        TableSchema schema = catalog.getTable(p.table);

        // 打开事务管理器和数据管理器
        TransactionManager tm = TransactionManager.open("data/" + p.table + "_tm");
        DataManager dm = DataManager.open("data/" + p.table + "_dm", 1<<20, tm);

        // 创建分页表堆
        PagedTableHeap heap = new PagedTableHeap(schema, tm, dm);

        // 构建过滤条件（如果有）
        Predicate<Map<String,Object>> filter = null;
        if (p.predicate != null) {
            filter = rec -> evaluateExpression(p.predicate, rec);
        }

        // 执行更新操作
        int cnt = heap.updateIf(filter, p.set);

        // 关闭资源
        dm.close(); tm.close();
        return cnt + " row(s) updated~";
    }

    /**
     * 执行删除操作
     * 目前只支持分页存储后端
     * @param p 删除计划
     * @return 删除影响的行数
     */
    private Object execDelete(DeletePlan p) {
        if (!usePaged()) return "DELETE not implemented for JSON heap";

        // 获取表结构信息
        TableSchema schema = catalog.getTable(p.table);

        // 打开事务管理器和数据管理器
        TransactionManager tm = TransactionManager.open("data/" + p.table + "_tm");
        DataManager dm = DataManager.open("data/" + p.table + "_dm", 1<<20, tm);

        // 创建分页表堆
        PagedTableHeap heap = new PagedTableHeap(schema, tm, dm);

        // 构建过滤条件（如果有）
        Predicate<Map<String,Object>> filter = null;
        if (p.predicate != null) {
            filter = rec -> evaluateExpression(p.predicate, rec);
        }

        // 执行删除操作
        int cnt = heap.deleteIf(filter);

        // 关闭资源
        dm.close(); tm.close();
        return cnt + " row(s) deleted~";
    }
} 